package com.hp.provider;


import com.alibaba.fastjson2.JSON;
import com.hp.provider.source.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.http.MediaType;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication
//@EnableBinding({PropertiesSource.class})
@RestController
public class SpringKafkaProviderApplication {

    //@Autowired
    //private PropertiesSource propertiesSource;

    @Autowired
    private StreamBridge streamBridge;


    public static void main(String[] args) {
        SpringApplication.run(SpringKafkaProviderApplication.class, args);
    }

    @GetMapping("/senMsg")
    public void senMsg() throws Exception {
        for (int i = 0; i < 10; i++) {
//                send("王五"+i);
//                send(User.builder().name("王五").age(i).build());
            send(JSON.toJSONString(User.builder().name("王五").age(i).build()));
            send(User.builder().name("李六").age(i).build());
            sendRabbit0(User.builder().name("王五").age(i).build());
            sendRabbit1(User.builder().name("李六").age(i).build());
        }
    }

    public void send(String msg) throws Exception {
        //boolean send = propertiesSource.sendMessage().send(MessageBuilder.withPayload(msg)
        //        .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
        //        .build());

        boolean send = streamBridge.send("outputKafka1-out-0", MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build());

        System.out.println("---------" + send);

    }


    public void send(User msg) throws Exception {
        boolean send = streamBridge.send("outputKafka2-out-0", MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build());

        System.out.println("---------" + send);

    }


    public void sendRabbit0(User msg) throws Exception {
        boolean send = streamBridge.send("outputRabbit1-out-0", MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build());
        System.out.println("---------" + send);

    }

    public void sendRabbit1(User msg) throws Exception {
        boolean send = streamBridge.send("outputRabbit2-out-0", MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build());

        System.out.println("---------" + send);

    }


//    public <T> void sendWithTags(T msg, String tag) throws Exception {
//        Message message = MessageBuilder.createMessage(msg,
//                new MessageHeaders(Stream.of(tag).collect(Collectors
//                        .toMap(str -> MessageConst.PROPERTY_TAGS, String::toString))));
//        propertiesSource.sendMessage().send(message);
//    }
//
//    public <T> void sendWithKeys(T msg, String key) throws Exception {
//        Message message = MessageBuilder.createMessage(msg,
//                new MessageHeaders(Stream.of(key).collect(Collectors
//                        .toMap(str -> MessageConst.PROPERTY_KEYS, String::toString))));
//        propertiesSource.sendMessage().send(message);
//    }
//
//    public <T> void sendObject(T msg, String tag) throws Exception {
//        Message message = MessageBuilder.withPayload(msg)
//                .setHeader(MessageConst.PROPERTY_TAGS, tag)
//                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
//                .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL,1)
//                .build();
//        propertiesSource.sendMessage().send(message);
//    }

//    public <T> void sendTransactionalMsg(T msg, int num) throws Exception {
//        MessageBuilder builder = MessageBuilder.withPayload(msg)
//                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
//        builder.setHeader("test", String.valueOf(num));
//        builder.setHeader(RocketMQHeaders.TAGS, "binder");
//        Message message = builder.build();
//        propertiesSource.output2().send(message);
//    }

}
